print 'onVirq>', val
self.refresh()
- def schedule_later(self, _delay, _name, _fn, *args):
- """Schedule a function to be called later (if not already scheduled).
-
- @param _delay: delay in seconds
- @param _name: schedule name
- @param _fn: function
- @param args: arguments
- """
- self.scheduler.later(_delay, _name, _fn, args)
-
- def schedule_cancel(self, name):
- """Cancel a scheduled function call.
-
- @param name: schedule name to cancel
- """
- self.scheduler.cancel(name)
-
- def domain_restarts_schedule(self, delay=1):
- """Schedule domain_restarts to be called later.
-
- @param delay: delay in seconds
- """
- self.schedule_later(delay, 'domain_restarts', self.domain_restarts)
-
- def domain_restarts_cancel(self):
- """Cancel any scheduled call of domain_restarts.
- """
- self.schedule_cancel('domain_restarts')
-
def rm_all(self):
"""Remove all domain info. Used after reboot.
"""
eserver.inject('xend.domain.exit', [name, id, 'crash'])
destroyed += 1
self.final_domain_destroy(id)
- if self.domain_restarts_exist():
- self.domain_restarts_schedule()
def refresh(self):
"""Refresh domain list from Xen.
self.domain_lookup(id)
# Remove entries for domains that no longer exist.
# Update entries for existing domains.
+ do_domain_restarts = False
for d in self.domain_by_id.values():
info = doms.get(d.id)
if info:
d.update(info)
elif d.restart_pending():
- pass
+ do_domain_restarts = True
else:
self._delete_domain(d.id)
+ if do_domain_restarts:
+ self.scheduler.now(self.domain_restarts)
def update_domain(self, id):
"""Update the saved info for a domain.
@return: domain names
"""
- self.refresh()
return self.domain_by_id.keys()
def domains(self):
@return: domain objects
"""
- self.refresh()
return self.domain_by_id.values()
def domain_create(self, config):
"""Process pending domain shutdowns.
Destroys domains whose shutdowns have timed out.
"""
- self.schedule_cancel('domain_shutdowns')
timeout = SHUTDOWN_TIMEOUT
for shutdown in self.shutdowns_by_id.values():
id = shutdown.getDomain()
timeout = min(timeout, shutdown.getTimeout())
if self.shutdowns_by_id:
# Pending shutdowns remain - reschedule.
- self.schedule_later(timeout, 'domain_shutdowns', self.domain_shutdowns)
+ self.scheduler.later(timeout, self.domain_shutdowns)
def domain_restart_schedule(self, id, reason, force=False):
"""Schedule a restart for a domain if it needs one.
log.info('Scheduling restart for domain: name=%s id=%s', dominfo.name, dominfo.id)
eserver.inject("xend.domain.restart",
[dominfo.name, dominfo.id, "schedule"])
- self.domain_restarts_schedule()
def domain_restart_cancel(self, id):
"""Cancel any restart scheduled for a domain.
def domain_restarts(self):
"""Execute any scheduled domain restarts for domains that have gone.
"""
- self.domain_restarts_cancel()
doms = self.xen_domains()
for dominfo in self.restarts_by_id.values():
print 'domain_restarts>', dominfo.name, dominfo.id
del self.restarts_by_name[dominfo.name]
print 'domain_restarts> restarting: ', dominfo.name
self.domain_restart(dominfo)
- if self.domain_restarts_exist():
- # Run again later if any restarts remain.
- self.domain_restarts_schedule(delay=10)
- def domain_restarts_exist(self):
- return len(self.restarts_by_id)
-
def final_domain_destroy(self, id):
"""Final destruction of a domain..
"""
self.domain_restart_schedule(id, reason, force=True)
val = self.final_domain_destroy(id)
- self.refresh()
return val
def domain_migrate(self, id, dst, live=False, resource=0):
class Scheduler:
- def __init__(self):
- self.lock = threading.Lock()
- self.schedule = {}
-
- def later(self, _delay, _name, _fn, args, kwargs={}):
- """Schedule a function to be called later (if not already scheduled).
+ def later(self, _delay, _fn, args=(), kwargs={}):
+ """Schedule a function to be called later.
@param _delay: delay in seconds
- @param _name: schedule name
@param _fn: function
@param args: arguments (list)
@param kwargs keyword arguments (map)
"""
- try:
- self.lock.acquire()
- if self.schedule.get(_name): return
- runargs = [ _name, _fn, args, kwargs ]
- timer = threading.Timer(_delay, self._run, args=runargs)
- self.schedule[_name] = timer
- finally:
- self.lock.release()
+ runargs = [ _fn, args, kwargs ]
+ timer = threading.Timer(_delay, self._run, args=runargs)
timer.start()
- def cancel(self, name):
- """Cancel a scheduled function call.
-
- @param name: schedule name to cancel
- """
- timer = self._remove(name)
- if timer:
- timer.cancel()
+ def now(self, _fn, args=(), kwargs={}):
+ """Schedule a function to be called now.
- def _remove(self, name):
- try:
- self.lock.acquire()
- timer = self.schedule.get(name)
- if timer:
- del self.schedule[name]
- return timer
- finally:
- self.lock.release()
+ @param _fn: function
+ @param args: arguments (list)
+ @param kwargs keyword arguments (map)
+ """
+ runargs = [ _fn, args, kwargs ]
+ thread = threading.Thread(target=self._run, args=runargs)
+ thread.start()
- def _run(self, name, fn, args, kwargs):
- self._remove(name)
+ def _run(self, fn, args, kwargs):
fn(*args, **kwargs)
-
-
projects / xen.git / commitdiff